package com.ideal.tools.scheduler;

import com.ideal.tools.db.MysqlDBUtils;
import com.ideal.tools.ssh.common.CommonProperties;
import com.ideal.tools.ssh.common.CommonTools;
import com.ideal.tools.ssh.entity.FairScheduler;
import org.apache.commons.lang3.StringUtils;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.transform.OutputKeys;
import javax.xml.transform.Transformer;
import javax.xml.transform.TransformerException;
import javax.xml.transform.TransformerFactory;
import javax.xml.transform.dom.DOMSource;
import javax.xml.transform.stream.StreamResult;
import java.io.*;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;

/**
 * Created by CC on 2016/3/30.
 */
public class DbFairSchedulerTools {


    public static String FAIR_SCHEDULER_QUEUE_LIST = "FAIR_SCHEDULER_QUEUE_LIST";


    /**
     *
     * @param commonProperties
     * @param documentBuilder
     * @throws IOException
     * @throws TransformerException
     */
    public static String writeFairScheduleFileToLocal(
            CommonProperties commonProperties, DocumentBuilder documentBuilder)
            throws IOException, TransformerException {
        //公平调度文件名称
        String fairFileName = commonProperties.
                getProperty(CommonProperties.RM_SCHEDULE_XML_FILE_NAME, "");
        //创建出一个带时间戳的 文件名 这样可以追溯历史
        String timeFileName= CommonTools.getTimeFileName(fairFileName);

        String finalFileName =  commonProperties.
                getProperty(CommonProperties.RM_SCHEDULE_XML_SRC_PATH, "")+timeFileName;

        //本机文件
        File localFile = new File(finalFileName);
        if(!localFile.exists()){
            localFile.createNewFile();
        }

        Document document = documentBuilder.newDocument();

        //把队列信息转换成xml
        translateXML(commonProperties,document);

        //把xml写入文件
        writeXML2File(document,localFile);

        return finalFileName;
    }



    private static void translateXML(CommonProperties commonProperties, Document document){
        Element root = document.createElement("allocations");
        //root queue

        Element rootQueue = document.createElement("queue");
        rootQueue.setAttribute("name", "root");
        String rootAcl = commonProperties.getProperty(CommonProperties.FAIR_SCHEDULER_ROOT_ACL,"hadoop supergroup");

        Element aclSubmitApps = document.createElement("aclSubmitApps");
        aclSubmitApps.setTextContent(rootAcl);
        rootQueue.appendChild(aclSubmitApps);
        Element aclAdministerApps = document.createElement("aclAdministerApps");
        aclAdministerApps.setTextContent(rootAcl);
        rootQueue.appendChild(aclAdministerApps);

        //这个list 应该是从前台传递过来的 这里不再自己去取
//        List<FairScheduler> fairSchedulerQueues = (List<FairScheduler>)
//                commonProperties.getObjectParamter(FAIR_SCHEDULER_QUEUE_LIST,null);
        List<FairScheduler> fairSchedulerQueues = DbFairSchedulerTools.getFairSchedulerList();
        if(fairSchedulerQueues==null)
            return;
        //处理原始的 fairSchedule
        processOriginalFairSchedule(fairSchedulerQueues);

        ListIterator<FairScheduler> it  = fairSchedulerQueues.listIterator();
        while(it.hasNext()){
            FairScheduler fairSchedulerQueue = it.next();

            /*配置队列名*/
            Element queue = document.createElement("queue");
            queue.setAttribute("name", fairSchedulerQueue.getQueue());

            /*配置最小资源量*/
            StringBuffer minResourcesTmp = new StringBuffer();
            minResourcesTmp.append(fairSchedulerQueue.getMinMemory()).append(" ").append("mb, ");
            minResourcesTmp.append(fairSchedulerQueue.getMinCpu()).append(" ").append("vcores");
            String minResourcess =  minResourcesTmp.toString();
            Element minResources = document.createElement("minResources");
            minResources.setTextContent(minResourcess);
            queue.appendChild(minResources);
            /*配置最大资源量*/
            StringBuffer maxResourcesTmp = new StringBuffer();
            maxResourcesTmp.append(fairSchedulerQueue.getMaxMemory()).append(" ").append("mb, ");
            maxResourcesTmp.append(fairSchedulerQueue.getMaxCpu()).append(" ").append("vcores");
            String maxResourcess  =  maxResourcesTmp.toString();
            Element maxResources = document.createElement("maxResources");
            maxResources.setTextContent(maxResourcess);
            queue.appendChild(maxResources);

            /*配置用户能运行的最大任务数*/
            if (fairSchedulerQueue.getMaxApps() != null) {
                Element maxRunningApps = document.createElement("maxRunningApps");
                maxRunningApps.setTextContent(String.valueOf(fairSchedulerQueue.getMaxApps()));
                queue.appendChild(maxRunningApps);
            }
            /*配置用户权重*/
            if (fairSchedulerQueue.getWeight() != null) {
                Element weightElement = document.createElement("weight");
                weightElement.setTextContent(String.valueOf(fairSchedulerQueue.getWeight()));
                queue.appendChild(weightElement);
            }
            //acl control
            String acl = fairSchedulerQueue.getUsername();
            if (StringUtils.isNotEmpty(fairSchedulerQueue.getAcl())) {
                acl = fairSchedulerQueue.getAcl();
            }
            /*aclSubmitApps*/
            Element userAclSubmitApps = document.createElement("aclSubmitApps");
            userAclSubmitApps.setTextContent(acl);
            queue.appendChild(userAclSubmitApps);
            /*aclAdministerApps*/
            Element userAclAdministerApps = document.createElement("aclAdministerApps");
            userAclAdministerApps.setTextContent(acl);
            queue.appendChild(userAclAdministerApps);

            rootQueue.appendChild(queue);
        }
        root.appendChild(rootQueue);

        document.appendChild(root);


    }

    /**
     * 对原始的队列集合做处理
     * 这里只用是在原始中加入 一个默认队列
     * @param fairSchedulerQueues
     */
    private static void processOriginalFairSchedule(List<FairScheduler> fairSchedulerQueues){
        FairScheduler fairSchedulerQueue = new FairScheduler();
        fairSchedulerQueue.setUsername("default");
        fairSchedulerQueue.setQueue("default");
        fairSchedulerQueue.setAcl("*");
        fairSchedulerQueue.setId(0l);
        fairSchedulerQueue.setMinCpu(25);
        fairSchedulerQueue.setMaxCpu(100);
        fairSchedulerQueue.setMinMemory(102400);
        fairSchedulerQueue.setMaxMemory(409600);
        fairSchedulerQueue.setMaxApps(50);
        fairSchedulerQueue.setWeight(Float.valueOf(1));
        //加入集合
        fairSchedulerQueues.add(fairSchedulerQueue);
    }


    private static void writeXML2File(Document doc, File file) throws TransformerException, IOException {
        TransformerFactory factory = TransformerFactory.newInstance();
        factory.setAttribute("indent-number", new Integer(4));// 设置缩进长度为4
        Transformer transformer = factory.newTransformer();
        transformer.setOutputProperty(OutputKeys.INDENT, "yes");// 设置自动换行
        doc.normalize();
        DOMSource source = new DOMSource(doc);
        Writer out = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file), "UTF-8"));
        transformer.transform(source, new StreamResult(out));
        out.flush();
        out.close();
    }


    public static List<FairScheduler> getFairSchedulerList(){
        List<FairScheduler> fairSchedulerList= new ArrayList<FairScheduler>();
        String sql="select cluster_user_queue.id id,cluster_user.userName username,cluster_user_queue.queueName queuename,cluster_user_queue.minResource minresource," +
                "cluster_user_queue.maxResource maxresource,cluster_user_queue.maxApp maxapp,cluster_user_queue.weight weight," +
                "cluster_user_queue.acl acl,cluster_user_queue.minCpu mincpu,cluster_user_queue.maxcpu maxcpu " +
                "from cluster_user_queue left join cluster_user on cluster_user_queue.userId = cluster_user.id";
        List<Map<String,Object>> rows=MysqlDBUtils.queryWebDB(sql);
        for (Map<String,Object> row:rows){
            FairScheduler fairSchedulerQueue= new FairScheduler();
            fairSchedulerQueue.setUsername(row.get("username").toString());
            fairSchedulerQueue.setQueue(row.get("queuename").toString());
            fairSchedulerQueue.setAcl(row.get("acl").toString());
            fairSchedulerQueue.setId(Long.parseLong(row.get("id").toString()));
            fairSchedulerQueue.setMinCpu(Integer.parseInt(row.get("mincpu").toString()));
            fairSchedulerQueue.setMaxCpu(Integer.parseInt(row.get("maxcpu").toString()));
            fairSchedulerQueue.setMinMemory(Integer.parseInt(row.get("minresource").toString()));
            fairSchedulerQueue.setMaxMemory(Integer.parseInt(row.get("maxresource").toString()));
            fairSchedulerQueue.setMaxApps(Integer.parseInt(row.get("maxapp").toString()));
            fairSchedulerQueue.setWeight(Float.parseFloat(row.get("weight").toString()));
            fairSchedulerList.add(fairSchedulerQueue);
        }

        return fairSchedulerList;
    }


}
